We now pass in rdbSnapshotOptions options in this function, and
options.conns
is now malloc'ed in the caller side, so we need to zfree it when
returning early
due to an error. Previously, conns was malloc'ed after the error
handling, so we
don't have this.

Introduced in #1949.

---------

Signed-off-by: Binbin <[email protected]>

Introduces a new family of commands for migrating slots via replication.
The procedure is driven by the source node which pushes an AOF formatted
snapshot of the slots to the target, followed by a replication stream of
changes on that slot (a la manual failover).

This solution is an adaptation of the solution provided by
@enjoy-binbin, combined with the solution I previously posted at valkey-io#1591,
modified to meet the designs we had outlined in valkey-io#23.

## New commands

* `CLUSTER MIGRATESLOTS SLOTSRANGE start end [start end]... NODE
node-id`: Begin sending the slot via replication to the target. Multiple
targets can be specified by repeating `SLOTSRANGE ... NODE ...`
*  `CLUSTER CANCELMIGRATION ALL`: Cancel all slot migrations
* `CLUSTER GETSLOTMIGRATIONS`: See a recent log of migrations

This PR only implements "one shot" semantics with an asynchronous model.
Later, "two phase" (e.g. slot level replicate/failover commands) can be
added with the same core.

## Slot migration jobs

Introduces the concept of a slot migration job. While active, a job
tracks a connection created by the source to the target over which the
contents of the slots are sent. This connection is used for control
messages as well as replicated slot data. Each job is given a 40
character random name to help uniquely identify it.

All jobs, including those that finished recently, can be observed using
the `CLUSTER GETSLOTMIGRATIONS` command.

## Replication

* Since the snapshot uses AOF, the snapshot can be replayed verbatim to
any replicas of the target node.
* We use the same proxying mechanism used for chaining replication to
copy the content sent by the source node directly to the replica nodes.

## `CLUSTER SYNCSLOTS`

To coordinate the state machine transitions across the two nodes, a new
command is added, `CLUSTER SYNCSLOTS`, that performs this control flow.

Each end of the slot migration connection is expected to install a read
handler in order to handle `CLUSTER SYNCSLOTS` commands:

* `ESTABLISH`: Begins a slot migration. Provides slot migration
information to the target and authorizes the connection to write to
unowned slots.
* `SNAPSHOT-EOF`: appended to the end of the snapshot to signal that the
snapshot is done being written to the target.
* `PAUSE`: informs the source node to pause whenever it gets the
opportunity
* `PAUSED`: added to the end of the client output buffer when the pause
is performed. The pause is only performed after the buffer shrinks below
a configurable size
* `REQUEST-FAILOVER`: request the source to either grant or deny a
failover for the slot migration. The grant is only granted if the target
is still paused. Once a failover is granted, the paused is refreshed for
a short duration
* `FAILOVER-GRANTED`: sent to the target to inform that REQUEST-FAILOVER
is granted
* `ACK`: heartbeat command used to ensure liveness

## Interactions with other commands

* FLUSHDB on the source node (which flushes the migrating slot) will
result in the source dropping the connection, which will flush the slot
on the target and reset the state machine back to the beginning. The
subsequent retry should very quickly succeed (it is now empty)
* FLUSHDB on the target will fail the slot migration. We can iterate
with better handling, but for now it is expected that the operator would
retry.
* Genearlly, FLUSHDB is expected to be executed cluster wide, so
preserving partially migrated slots doesn't make much sense
* SCAN and KEYS are filtered to avoid exposing importing slot data

## Error handling

* For any transient connection drops, the migration will be failed and
require the user to retry.
* If there is an OOM while reading from the import connection, we will
fail the import, which will drop the importing slot data
* If there is a client output buffer limit reached on the source node,
it will drop the connection, which will cause the migration to fail
* If at any point the export loses ownership or either node is failed
over, a callback will be triggered on both ends of the migration to fail
the import. The import will not reattempt with a new owner
* The two ends of the migration are routinely pinging each other with
SYNCSLOTS ACK messages. If at any point there is no interaction on the
connection for longer than `repl-timeout`, the connection will be
dropped, resulting in migration failure
* If a failover happens, we will drop keys in all unowned slots. The
migration does not persist through failovers and would need to be
retried on the new source/target.

## State machine

```
                                                                            
                Target/Importing Node State Machine                         
   ─────────────────────────────────────────────────────────────            
                                                                            
             ┌────────────────────┐
             │SLOT_IMPORT_WAIT_ACK┼──────┐
             └──────────┬─────────┘      │
                     ACK│                │
         ┌──────────────▼─────────────┐  │
         │SLOT_IMPORT_RECEIVE_SNAPSHOT┼──┤
         └──────────────┬─────────────┘  │
            SNAPSHOT-EOF│                │                                  
        ┌───────────────▼──────────────┐ │                                  
        │SLOT_IMPORT_WAITING_FOR_PAUSED┼─┤                                  
        └───────────────┬──────────────┘ │                                  
                  PAUSED│                │                                  
        ┌───────────────▼──────────────┐ │ Error Conditions:                
        │SLOT_IMPORT_FAILOVER_REQUESTED┼─┤  1. OOM                          
        └───────────────┬──────────────┘ │  2. Slot Ownership Change        
        FAILOVER-GRANTED│                │  3. Demotion to replica          
         ┌──────────────▼─────────────┐  │  4. FLUSHDB                      
         │SLOT_IMPORT_FAILOVER_GRANTED┼──┤  5. Connection Lost              
         └──────────────┬─────────────┘  │  6. No ACK from source (timeout) 
      Takeover Performed│                │                                  
         ┌──────────────▼───────────┐    │                                  
         │SLOT_MIGRATION_JOB_SUCCESS┼────┤                                  
         └──────────────────────────┘    │                                  
                                         │                                  
   ┌─────────────────────────────────────▼─┐                                
   │SLOT_IMPORT_FINISHED_WAITING_TO_CLEANUP│                                
   └────────────────────┬──────────────────┘                                
Unowned Slots Cleaned Up│                                                   
          ┌─────────────▼───────────┐                                      
          │SLOT_MIGRATION_JOB_FAILED│                                      
          └─────────────────────────┘                                      

                                                                                           
                                                                                           
                      Source/Exporting Node State Machine                                  
         ─────────────────────────────────────────────────────────────                     
                                                                                           
               ┌──────────────────────┐                                                    
               │SLOT_EXPORT_CONNECTING├─────────┐                                          
               └───────────┬──────────┘         │                                          
                  Connected│                    │                                          
             ┌─────────────▼────────────┐       │                                          
             │SLOT_EXPORT_AUTHENTICATING┼───────┤                                          
             └─────────────┬────────────┘       │                                          
              Authenticated│                    │                                          
             ┌─────────────▼────────────┐       │                                          
             │SLOT_EXPORT_SEND_ESTABLISH┼───────┤                                          
             └─────────────┬────────────┘       │                                          
  ESTABLISH command written│                    │                                          
     ┌─────────────────────▼─────────────┐      │                                          
     │SLOT_EXPORT_READ_ESTABLISH_RESPONSE┼──────┤                                          
     └─────────────────────┬─────────────┘      │                                          
   Full response read (+OK)│                    │                                          
          ┌────────────────▼──────────────┐     │ Error Conditions:                        
          │SLOT_EXPORT_WAITING_TO_SNAPSHOT┼─────┤  1. User sends CANCELMIGRATION           
          └────────────────┬──────────────┘     │  2. Slot ownership change                
     No other child process│                    │  3. Demotion to replica                  
              ┌────────────▼───────────┐        │  4. FLUSHDB                              
              │SLOT_EXPORT_SNAPSHOTTING┼────────┤  5. Connection Lost                      
              └────────────┬───────────┘        │  6. AUTH failed                          
              Snapshot done│                    │  7. ERR from ESTABLISH command           
               ┌───────────▼─────────┐          │  8. Unpaused before failover completed   
               │SLOT_EXPORT_STREAMING┼──────────┤  9. Snapshot failed (e.g. Child OOM)     
               └───────────┬─────────┘          │  10. No ack from target (timeout)        
                      PAUSE│                    │  11. Client output buffer overrun        
            ┌──────────────▼─────────────┐      │                                          
            │SLOT_EXPORT_WAITING_TO_PAUSE┼──────┤                                          
            └──────────────┬─────────────┘      │                                          
             Buffer drained│                    │                                          
            ┌──────────────▼────────────┐       │                                          
            │SLOT_EXPORT_FAILOVER_PAUSED┼───────┤                                          
            └──────────────┬────────────┘       │                                          
   Failover request granted│                    │                                          
           ┌───────────────▼────────────┐       │                                          
           │SLOT_EXPORT_FAILOVER_GRANTED┼───────┤                                          
           └───────────────┬────────────┘       │                                          
      New topology received│                    │                                          
            ┌──────────────▼───────────┐        │                                          
            │SLOT_MIGRATION_JOB_SUCCESS│        │                                          
            └──────────────────────────┘        │                                          
                                                │                                          
            ┌─────────────────────────┐         │                                          
            │SLOT_MIGRATION_JOB_FAILED│◄────────┤                                          
            └─────────────────────────┘         │                                          
                                                │                                          
           ┌────────────────────────────┐       │                                          
           │SLOT_MIGRATION_JOB_CANCELLED│◄──────┘                                          
           └────────────────────────────┘                                                 
```

Co-authored-by: Binbin <[email protected]>

---------

Signed-off-by: Binbin <[email protected]>
Signed-off-by: Jacob Murphy <[email protected]>
Signed-off-by: Madelyn Olson <[email protected]>
Co-authored-by: Binbin <[email protected]>
Co-authored-by: Ping Xie <[email protected]>
Co-authored-by: Madelyn Olson <[email protected]>

We now pass in rdbSnapshotOptions options in this function, and
options.conns
is now malloc'ed in the caller side, so we need to zfree it when
returning early
due to an error. Previously, conns was malloc'ed after the error
handling, so we
don't have this.

Introduced in valkey-io#1949.

---------

Signed-off-by: Binbin <[email protected]>

We now pass in rdbSnapshotOptions options in this function, and
options.conns
is now malloc'ed in the caller side, so we need to zfree it when
returning early
due to an error. Previously, conns was malloc'ed after the error
handling, so we
don't have this.

Introduced in valkey-io#1949.

---------

Signed-off-by: Binbin <[email protected]>

This may result in meaningless slot migration job, we should
return an error to user in advance to avoid operation error.
Also `by myself` is not correct English grammar and `myself`
is a internal code terminology, changed to `by this node`.

Was introduced in #1949.

---------

Signed-off-by: Binbin <[email protected]>

If all cluster nodes have functions, slot migration will fail
since the target will return the function already exists error
when doing the FUNCTION LOAD.

And in addition, the target's replica could panic when it executes
the FUNCTION LOAD propagated from the primary (see propagation-error-behavior).

Introduced in valkey-io#1949.

Signed-off-by: Binbin <[email protected]>

If all cluster nodes have functions, slot migration will fail
since the target will return the function already exists error
when doing the FUNCTION LOAD.

And in addition, the target's replica could panic when it executes
the FUNCTION LOAD propagated from the primary (see
propagation-error-behavior).

Introduced in #1949.

Signed-off-by: Binbin <[email protected]>

…ous reading of auth response (#2494)

The old SLOT_EXPORT_AUTHENTICATING added in #1949, when processed by the source node,
we will send the AUTH command and then reads the response. If the target node is blocked
during this process, the source node will also be blocked. We should use a read handler
to handle this. We split SLOT_EXPORT_AUTHENTICATING into SLOT_EXPORT_SEND_AUTH and
SLOT_EXPORT_READ_AUTH_RESPONSE to avoid this issue.

Signed-off-by: Binbin <[email protected]>

When we adding atomic slot migration in #1949, we reused a lot of rdb save code,
it was an easier way to implement ASM in the first time, but it comes with some
side effect. Like we are using CHILD_TYPE_RDB to do the fork, we use rdb.c/rdb.h
function to save the snapshot, these mess up the logs (we will print some logs
saying we are doing RDB stuff) and mess up the info fields (we will say we are
rdb_bgsave_in_progress but actually we are doing slot migration).

In addition, it makes the code difficult to maintain. The rdb_save method uses
a lot of rdb_* variables, but we are actually doing slot migration. If we want
to support one fork with multiple target nodes, we need to rewrite these code
for a better cleanup.

Note that the changes to rdb.c/rdb.h are reverting previous changes from when
we was reusing this code for slot migration. The slot migration snapshot logic
is similar to the previous diskless replication. We use pipe to transfer the
snapshot data from the child process to the parent process.

Interface changes:
- New slot_migration_fork_in_progress info field.
- New cow_size field in CLUSTER GETSLOTMIGRATIONS command.
- Also add slot migration fork to the cluster class trace latency.

Signed-off-by: Binbin <[email protected]>
Signed-off-by: Jacob Murphy <[email protected]>
Co-authored-by: Jacob Murphy <[email protected]>

We now pass in rdbSnapshotOptions options in this function, and
options.conns
is now malloc'ed in the caller side, so we need to zfree it when
returning early
due to an error. Previously, conns was malloc'ed after the error
handling, so we
don't have this.

Introduced in valkey-io#1949.

---------

Signed-off-by: Binbin <[email protected]>

This may result in meaningless slot migration job, we should
return an error to user in advance to avoid operation error.
Also `by myself` is not correct English grammar and `myself`
is a internal code terminology, changed to `by this node`.

Was introduced in valkey-io#1949.

---------

Signed-off-by: Binbin <[email protected]>

If all cluster nodes have functions, slot migration will fail
since the target will return the function already exists error
when doing the FUNCTION LOAD.

And in addition, the target's replica could panic when it executes
the FUNCTION LOAD propagated from the primary (see
propagation-error-behavior).

Introduced in valkey-io#1949.

Signed-off-by: Binbin <[email protected]>

…ous reading of auth response (valkey-io#2494)

The old SLOT_EXPORT_AUTHENTICATING added in valkey-io#1949, when processed by the source node,
we will send the AUTH command and then reads the response. If the target node is blocked
during this process, the source node will also be blocked. We should use a read handler
to handle this. We split SLOT_EXPORT_AUTHENTICATING into SLOT_EXPORT_SEND_AUTH and
SLOT_EXPORT_READ_AUTH_RESPONSE to avoid this issue.

Signed-off-by: Binbin <[email protected]>

When we adding atomic slot migration in valkey-io#1949, we reused a lot of rdb save code,
it was an easier way to implement ASM in the first time, but it comes with some
side effect. Like we are using CHILD_TYPE_RDB to do the fork, we use rdb.c/rdb.h
function to save the snapshot, these mess up the logs (we will print some logs
saying we are doing RDB stuff) and mess up the info fields (we will say we are
rdb_bgsave_in_progress but actually we are doing slot migration).

In addition, it makes the code difficult to maintain. The rdb_save method uses
a lot of rdb_* variables, but we are actually doing slot migration. If we want
to support one fork with multiple target nodes, we need to rewrite these code
for a better cleanup.

Note that the changes to rdb.c/rdb.h are reverting previous changes from when
we was reusing this code for slot migration. The slot migration snapshot logic
is similar to the previous diskless replication. We use pipe to transfer the
snapshot data from the child process to the parent process.

Interface changes:
- New slot_migration_fork_in_progress info field.
- New cow_size field in CLUSTER GETSLOTMIGRATIONS command.
- Also add slot migration fork to the cluster class trace latency.

Signed-off-by: Binbin <[email protected]>
Signed-off-by: Jacob Murphy <[email protected]>
Co-authored-by: Jacob Murphy <[email protected]>

We now pass in rdbSnapshotOptions options in this function, and
options.conns
is now malloc'ed in the caller side, so we need to zfree it when
returning early
due to an error. Previously, conns was malloc'ed after the error
handling, so we
don't have this.

Introduced in #1949.

---------

Signed-off-by: Binbin <[email protected]>

This may result in meaningless slot migration job, we should
return an error to user in advance to avoid operation error.
Also `by myself` is not correct English grammar and `myself`
is a internal code terminology, changed to `by this node`.

Was introduced in #1949.

---------

Signed-off-by: Binbin <[email protected]>

If all cluster nodes have functions, slot migration will fail
since the target will return the function already exists error
when doing the FUNCTION LOAD.

And in addition, the target's replica could panic when it executes
the FUNCTION LOAD propagated from the primary (see
propagation-error-behavior).

Introduced in #1949.

Signed-off-by: Binbin <[email protected]>

…ous reading of auth response (#2494)

The old SLOT_EXPORT_AUTHENTICATING added in #1949, when processed by the source node,
we will send the AUTH command and then reads the response. If the target node is blocked
during this process, the source node will also be blocked. We should use a read handler
to handle this. We split SLOT_EXPORT_AUTHENTICATING into SLOT_EXPORT_SEND_AUTH and
SLOT_EXPORT_READ_AUTH_RESPONSE to avoid this issue.

Signed-off-by: Binbin <[email protected]>

When we adding atomic slot migration in #1949, we reused a lot of rdb save code,
it was an easier way to implement ASM in the first time, but it comes with some
side effect. Like we are using CHILD_TYPE_RDB to do the fork, we use rdb.c/rdb.h
function to save the snapshot, these mess up the logs (we will print some logs
saying we are doing RDB stuff) and mess up the info fields (we will say we are
rdb_bgsave_in_progress but actually we are doing slot migration).

In addition, it makes the code difficult to maintain. The rdb_save method uses
a lot of rdb_* variables, but we are actually doing slot migration. If we want
to support one fork with multiple target nodes, we need to rewrite these code
for a better cleanup.

Note that the changes to rdb.c/rdb.h are reverting previous changes from when
we was reusing this code for slot migration. The slot migration snapshot logic
is similar to the previous diskless replication. We use pipe to transfer the
snapshot data from the child process to the parent process.

Interface changes:
- New slot_migration_fork_in_progress info field.
- New cow_size field in CLUSTER GETSLOTMIGRATIONS command.
- Also add slot migration fork to the cluster class trace latency.

Signed-off-by: Binbin <[email protected]>
Signed-off-by: Jacob Murphy <[email protected]>
Co-authored-by: Jacob Murphy <[email protected]>

Introduces a new family of commands for migrating slots via replication.
The procedure is driven by the source node which pushes an AOF formatted
snapshot of the slots to the target, followed by a replication stream of
changes on that slot (a la manual failover).

This solution is an adaptation of the solution provided by
@enjoy-binbin, combined with the solution I previously posted at valkey-io#1591,
modified to meet the designs we had outlined in valkey-io#23.

## New commands

* `CLUSTER MIGRATESLOTS SLOTSRANGE start end [start end]... NODE
node-id`: Begin sending the slot via replication to the target. Multiple
targets can be specified by repeating `SLOTSRANGE ... NODE ...`
*  `CLUSTER CANCELMIGRATION ALL`: Cancel all slot migrations
* `CLUSTER GETSLOTMIGRATIONS`: See a recent log of migrations

This PR only implements "one shot" semantics with an asynchronous model.
Later, "two phase" (e.g. slot level replicate/failover commands) can be
added with the same core.

## Slot migration jobs

Introduces the concept of a slot migration job. While active, a job
tracks a connection created by the source to the target over which the
contents of the slots are sent. This connection is used for control
messages as well as replicated slot data. Each job is given a 40
character random name to help uniquely identify it.

All jobs, including those that finished recently, can be observed using
the `CLUSTER GETSLOTMIGRATIONS` command.

## Replication

* Since the snapshot uses AOF, the snapshot can be replayed verbatim to
any replicas of the target node.
* We use the same proxying mechanism used for chaining replication to
copy the content sent by the source node directly to the replica nodes.

## `CLUSTER SYNCSLOTS`

To coordinate the state machine transitions across the two nodes, a new
command is added, `CLUSTER SYNCSLOTS`, that performs this control flow.

Each end of the slot migration connection is expected to install a read
handler in order to handle `CLUSTER SYNCSLOTS` commands:

* `ESTABLISH`: Begins a slot migration. Provides slot migration
information to the target and authorizes the connection to write to
unowned slots.
* `SNAPSHOT-EOF`: appended to the end of the snapshot to signal that the
snapshot is done being written to the target.
* `PAUSE`: informs the source node to pause whenever it gets the
opportunity
* `PAUSED`: added to the end of the client output buffer when the pause
is performed. The pause is only performed after the buffer shrinks below
a configurable size
* `REQUEST-FAILOVER`: request the source to either grant or deny a
failover for the slot migration. The grant is only granted if the target
is still paused. Once a failover is granted, the paused is refreshed for
a short duration
* `FAILOVER-GRANTED`: sent to the target to inform that REQUEST-FAILOVER
is granted
* `ACK`: heartbeat command used to ensure liveness

## Interactions with other commands

* FLUSHDB on the source node (which flushes the migrating slot) will
result in the source dropping the connection, which will flush the slot
on the target and reset the state machine back to the beginning. The
subsequent retry should very quickly succeed (it is now empty)
* FLUSHDB on the target will fail the slot migration. We can iterate
with better handling, but for now it is expected that the operator would
retry.
* Genearlly, FLUSHDB is expected to be executed cluster wide, so
preserving partially migrated slots doesn't make much sense
* SCAN and KEYS are filtered to avoid exposing importing slot data

## Error handling

* For any transient connection drops, the migration will be failed and
require the user to retry.
* If there is an OOM while reading from the import connection, we will
fail the import, which will drop the importing slot data
* If there is a client output buffer limit reached on the source node,
it will drop the connection, which will cause the migration to fail
* If at any point the export loses ownership or either node is failed
over, a callback will be triggered on both ends of the migration to fail
the import. The import will not reattempt with a new owner
* The two ends of the migration are routinely pinging each other with
SYNCSLOTS ACK messages. If at any point there is no interaction on the
connection for longer than `repl-timeout`, the connection will be
dropped, resulting in migration failure
* If a failover happens, we will drop keys in all unowned slots. The
migration does not persist through failovers and would need to be
retried on the new source/target.

## State machine

```

                Target/Importing Node State Machine
   ─────────────────────────────────────────────────────────────

             ┌────────────────────┐
             │SLOT_IMPORT_WAIT_ACK┼──────┐
             └──────────┬─────────┘      │
                     ACK│                │
         ┌──────────────▼─────────────┐  │
         │SLOT_IMPORT_RECEIVE_SNAPSHOT┼──┤
         └──────────────┬─────────────┘  │
            SNAPSHOT-EOF│                │
        ┌───────────────▼──────────────┐ │
        │SLOT_IMPORT_WAITING_FOR_PAUSED┼─┤
        └───────────────┬──────────────┘ │
                  PAUSED│                │
        ┌───────────────▼──────────────┐ │ Error Conditions:
        │SLOT_IMPORT_FAILOVER_REQUESTED┼─┤  1. OOM
        └───────────────┬──────────────┘ │  2. Slot Ownership Change
        FAILOVER-GRANTED│                │  3. Demotion to replica
         ┌──────────────▼─────────────┐  │  4. FLUSHDB
         │SLOT_IMPORT_FAILOVER_GRANTED┼──┤  5. Connection Lost
         └──────────────┬─────────────┘  │  6. No ACK from source (timeout)
      Takeover Performed│                │
         ┌──────────────▼───────────┐    │
         │SLOT_MIGRATION_JOB_SUCCESS┼────┤
         └──────────────────────────┘    │
                                         │
   ┌─────────────────────────────────────▼─┐
   │SLOT_IMPORT_FINISHED_WAITING_TO_CLEANUP│
   └────────────────────┬──────────────────┘
Unowned Slots Cleaned Up│
          ┌─────────────▼───────────┐
          │SLOT_MIGRATION_JOB_FAILED│
          └─────────────────────────┘

                      Source/Exporting Node State Machine
         ─────────────────────────────────────────────────────────────

               ┌──────────────────────┐
               │SLOT_EXPORT_CONNECTING├─────────┐
               └───────────┬──────────┘         │
                  Connected│                    │
             ┌─────────────▼────────────┐       │
             │SLOT_EXPORT_AUTHENTICATING┼───────┤
             └─────────────┬────────────┘       │
              Authenticated│                    │
             ┌─────────────▼────────────┐       │
             │SLOT_EXPORT_SEND_ESTABLISH┼───────┤
             └─────────────┬────────────┘       │
  ESTABLISH command written│                    │
     ┌─────────────────────▼─────────────┐      │
     │SLOT_EXPORT_READ_ESTABLISH_RESPONSE┼──────┤
     └─────────────────────┬─────────────┘      │
   Full response read (+OK)│                    │
          ┌────────────────▼──────────────┐     │ Error Conditions:
          │SLOT_EXPORT_WAITING_TO_SNAPSHOT┼─────┤  1. User sends CANCELMIGRATION
          └────────────────┬──────────────┘     │  2. Slot ownership change
     No other child process│                    │  3. Demotion to replica
              ┌────────────▼───────────┐        │  4. FLUSHDB
              │SLOT_EXPORT_SNAPSHOTTING┼────────┤  5. Connection Lost
              └────────────┬───────────┘        │  6. AUTH failed
              Snapshot done│                    │  7. ERR from ESTABLISH command
               ┌───────────▼─────────┐          │  8. Unpaused before failover completed
               │SLOT_EXPORT_STREAMING┼──────────┤  9. Snapshot failed (e.g. Child OOM)
               └───────────┬─────────┘          │  10. No ack from target (timeout)
                      PAUSE│                    │  11. Client output buffer overrun
            ┌──────────────▼─────────────┐      │
            │SLOT_EXPORT_WAITING_TO_PAUSE┼──────┤
            └──────────────┬─────────────┘      │
             Buffer drained│                    │
            ┌──────────────▼────────────┐       │
            │SLOT_EXPORT_FAILOVER_PAUSED┼───────┤
            └──────────────┬────────────┘       │
   Failover request granted│                    │
           ┌───────────────▼────────────┐       │
           │SLOT_EXPORT_FAILOVER_GRANTED┼───────┤
           └───────────────┬────────────┘       │
      New topology received│                    │
            ┌──────────────▼───────────┐        │
            │SLOT_MIGRATION_JOB_SUCCESS│        │
            └──────────────────────────┘        │
                                                │
            ┌─────────────────────────┐         │
            │SLOT_MIGRATION_JOB_FAILED│◄────────┤
            └─────────────────────────┘         │
                                                │
           ┌────────────────────────────┐       │
           │SLOT_MIGRATION_JOB_CANCELLED│◄──────┘
           └────────────────────────────┘
```

Co-authored-by: Binbin <[email protected]>

---------

Signed-off-by: Binbin <[email protected]>
Signed-off-by: Jacob Murphy <[email protected]>
Signed-off-by: Madelyn Olson <[email protected]>
Co-authored-by: Binbin <[email protected]>
Co-authored-by: Ping Xie <[email protected]>
Co-authored-by: Madelyn Olson <[email protected]>
Signed-off-by: Harkrishn Patro <[email protected]>

We now pass in rdbSnapshotOptions options in this function, and
options.conns
is now malloc'ed in the caller side, so we need to zfree it when
returning early
due to an error. Previously, conns was malloc'ed after the error
handling, so we
don't have this.

Introduced in valkey-io#1949.

---------

Signed-off-by: Binbin <[email protected]>
Signed-off-by: Harkrishn Patro <[email protected]>

This may result in meaningless slot migration job, we should
return an error to user in advance to avoid operation error.
Also `by myself` is not correct English grammar and `myself`
is a internal code terminology, changed to `by this node`.

Was introduced in valkey-io#1949.

---------

Signed-off-by: Binbin <[email protected]>
Signed-off-by: Harkrishn Patro <[email protected]>

If all cluster nodes have functions, slot migration will fail
since the target will return the function already exists error
when doing the FUNCTION LOAD.

And in addition, the target's replica could panic when it executes
the FUNCTION LOAD propagated from the primary (see
propagation-error-behavior).

Introduced in valkey-io#1949.

Signed-off-by: Binbin <[email protected]>
Signed-off-by: Harkrishn Patro <[email protected]>

…ous reading of auth response (valkey-io#2494)

The old SLOT_EXPORT_AUTHENTICATING added in valkey-io#1949, when processed by the source node,
we will send the AUTH command and then reads the response. If the target node is blocked
during this process, the source node will also be blocked. We should use a read handler
to handle this. We split SLOT_EXPORT_AUTHENTICATING into SLOT_EXPORT_SEND_AUTH and
SLOT_EXPORT_READ_AUTH_RESPONSE to avoid this issue.

Signed-off-by: Binbin <[email protected]>
Signed-off-by: Harkrishn Patro <[email protected]>